package cn.caplike.demo.project.reactor.quickstart;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

/**
 * Description: Threading and Schedulers<br>
 * Reference: 4.5.x <a href="https://projectreactor.io/docs/core/release/reference/#schedulers">Threading and Schedulers</a>
 *
 * @author LiKe
 * @version 1.0.0
 * @date 2020-08-20 20:12
 */
public class ThreadingAndSchedulers {

    /**
     * Description: 获取到一个 {@link Flux} 或是 {@link Mono} 并不意味着她们会运行在单独的线程里. 相反, 大多数运算符都工作在前一个操作符执行的线程里, 除非显示指定.
     *
     * @return void
     * @author LiKe
     * @date 2020-08-21 10:50:50
     */
    @Test
    public void testMonoRunInNewThread() throws InterruptedException {
        final Mono<String> mono = Mono.just("hello ");

        Thread t = new Thread(() -> mono
                .map(msg -> msg + "thread ")
                .subscribe(v ->
                        System.out.println(v + Thread.currentThread().getName())
                )
        );
        t.start();
        t.join();

        // 运行在主线程
        mono.map(msg -> msg + "thread ")
                .subscribe(v ->
                        System.out.println(v + Thread.currentThread().getName()));
    }

    // ~ Scheduler
    //   在 Project Reactor 中, 执行模型和什么时候执行是由它使用的 Scheduler 决定的. Scheduler 的职责和 ExecutorService 类似, 但是能做到更多.

    // =================================================================================================================

    // Project Reactor 提供了在响应式链中切换执行上下文的两种方式: publishOn 和 subscribeOn. 两个方法都需要一个 Scheduler 并需要开发者将执行上下文切换到这个 Scheduler 上.
    // But the placement of publishOn in the chain matters, while the placement of subscribeOn does not. To understand that difference,
    // you first have to remember that nothing happens until you subscribe.

    // In Reactor, when you chain operators, you can wrap as many Flux and Mono implementations inside one another as you need.
    // Once you subscribe, a chain of Subscriber objects is created, backward (up the chain) to the first publisher.
    // This is effectively hidden from you. All you can see is the outer layer of Flux (or Mono) and Subscription,
    // but these intermediate operator-specific subscribers are where the real work happens.

    // ~ 4.5.1 The publishOn Method
    //   https://projectreactor.io/docs/core/release/reference/#_the_publishon_method

    /**
     * Description: {@code publishOn} 的使用方式和其他运算符一致: 都是在订阅链的中段.
     *
     * @return void
     * @author LiKe
     * @date 2020-08-21 16:00:39
     */
    @Test
    public void testPublishOn() throws InterruptedException {
        // <1>.Creates a new Scheduler backed by four Thread instances.
        Scheduler scheduler = Schedulers.newParallel("parallel-scheduler", 4);

        final Flux<String> flux = Flux
                .range(1, 2)
                // <2>.The first map runs on the anonymous thread in <5>.
                .map(i -> {
                    System.out.println("The first map: " + Thread.currentThread().getName());
                    return 10 + i;
                })
                // <3>.The publishOn switches the whole sequence on a Thread picked from <1>.
                .publishOn(scheduler)
                // <4>.The second map runs on the Thread from <1>.
                .map(i -> {
                    System.out.println("The second map: " + Thread.currentThread().getName());
                    return "Result value: " + i;
                });

        // <5>.This anonymous Thread is the one where the subscription happens. The print happens on the latest execution context, which is the one from publishOn.
        new Thread(() -> {
            System.out.println("Start on " + Thread.currentThread().getName());
            flux.subscribe(System.out::println);
        }).start();
        // flux.subscribe(System.out::println);

        // Note that publishOn executes on a Scheduler. If we want to run that example in a main class,
        // we would need to add a sleep call here so that the application does not exit immediately without any value being produced
        Thread.sleep(2100);
    }

    // ~ 4.5.2 The subscribeOn Method
    //   https://projectreactor.io/docs/core/release/reference/#_the_subscribeon_method
    //   Changes the Thread from which the whole chain of operators subscribes
    //   Picks one thread from the Scheduler

    @Test
    public void testSubscribeOn() throws InterruptedException {
        // <1>.Creates a new Scheduler backed by four Thread.
        Scheduler scheduler = Schedulers.newParallel("parallel-scheduler", 4);

        final Flux<String> flux = Flux
                .range(1, 2)
                // <2>.The first map runs on one of these four threads…​
                .map(i -> {
                    System.out.println("The second map: " + Thread.currentThread().getName());
                    return 10 + i;
                })
                // <3>.…​because subscribeOn switches the whole sequence right from subscription time (<5>).
                .subscribeOn(scheduler)
                // <4>.>The second map also runs on same thread.
                .map(i -> {
                    System.out.println("The second map: " + Thread.currentThread().getName());
                    return "value " + i;
                });

        // <5>.This anonymous Thread is the one where the subscription initially happens, but subscribeOn immediately shifts it to one of the four scheduler threads.
        new Thread(() -> {
            System.out.println("Start on " + Thread.currentThread().getName());
            flux.subscribe(System.out::println);
        }).start();
        // flux.subscribe(System.out::println);
        // scheduler.schedule(task);

        // Note that subscribeOn executes on a Scheduler. If we want to run that example in a main class,
        // we would need to add a sleep call here so that the application does not exit immediately without any value being produced
        Thread.sleep(2100);
    }

}
